package com.stable.actor.core;

import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;

/**
 * 稳定的Actor基类 - 线程安全、资源可控
 */
@Slf4j
public abstract class StableActor implements Serializable {
    private static final long serialVersionUID = 1L;

    private final BlockingQueue<ActorMessage> mailbox = new LinkedBlockingQueue<>();
    private final ExecutorService dispatcher = Executors.newSingleThreadExecutor();
    protected  Map<Class<?>, Consumer<Object>> messageHandlers = new HashMap<>();
    private volatile boolean running = false;
    private ActorRef self;
    private ActorRef sender;
    private ActorSystem system;
    private ThreadLocal<ActorContext> currentContext = new ThreadLocal<>();

    public StableActor() {
        setupReceive();
    }

    // 初始化消息处理器
    private void setupReceive() {
        receiveBuilder()
                .match(String.class, this::handleString)
                .match(Heartbeat.class, this::handleHeartbeat)
                .matchAny(this::unhandled);

        // 子类可以添加更多处理器
        setupMessageHandlers();
    }

    // 子类重写此方法来定义消息处理
    protected abstract void setupMessageHandlers();

    // 启动Actor
    public void start() {
        if (!running) {
            running = true;
            dispatcher.submit(this::processMessages);
            preStart();
            onStart();
            log.info("✅ Actor启动: {}", getSelf().path());
        }
    }

    // 停止Actor
    public void stop() {
        running = false;
        dispatcher.shutdown();
        postStop();
        onStop();
        log.info("🛑 Actor停止: {}", getSelf().path());
    }

    // 消息处理循环
    private void processMessages() {
        while (running) {
            try {
                ActorMessage message = mailbox.take();
                log.info("从队列取出消息类型: {}", message.message.getClass().getName());
                log.info("从队列取出消息内容: {}", message.message);
                this.sender = message.sender;

                // 创建ActorContext并处理消息
                ActorContext context = new ActorContext(system, self, sender);
                currentContext.set(context); // 设置当前线程的上下文

                try {
                    handleMessageWithContext(message.message, context);
                } finally {
                    currentContext.remove(); // 清理ThreadLocal
                }

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("❌ Actor处理消息异常: {}", e.getMessage(), e);
                onError(e, null);
            } finally {
                // 重置 sender
                this.sender = null;
            }
        }
    }

    // 带Context的消息处理
    protected void handleMessageWithContext(Object message, ActorContext context) {
        Consumer<Object> handler = messageHandlers.get(message.getClass());
        if (handler != null) {
            handler.accept(message);
        } else {
            unhandled(message);
        }
    }

    // 传统的消息处理（保持兼容）
    protected void handleMessage(Object message) {
        Consumer<Object> handler = messageHandlers.get(message.getClass());
        if (handler != null) {
            handler.accept(message);
        } else {
            unhandled(message);
        }
    }

    /**
     * 获取当前消息处理的上下文
     * 只能在消息处理过程中调用
     */
    protected ActorContext getContext() {
        ActorContext context = currentContext.get();
        if (context == null) {
            throw new IllegalStateException("ActorContext is only available during message processing");
        }
        return context;
    }

    // 发送消息到本Actor
    public void tell(Object message, ActorRef sender) {
        if (running) {
            log.info("放入队列前消息类型: {}", message.getClass().getName());
            log.info("放入队列前消息内容: {}", message);
            mailbox.offer(new ActorMessage(message, sender));
        }
    }

    // 接收构建器 - 类似Akka API
    protected ReceiveBuilder receiveBuilder() {
        return new ReceiveBuilder();
    }

    // 内置消息处理器
    private void handleString(String message) {
        log.info("📝 收到字符串消息: {}", message);
    }

    private void handleHeartbeat(Heartbeat heartbeat) {
        log.info("💓 心跳来自: {}", heartbeat.from);
    }

    protected void unhandled(Object message) {
        log.warn("❓ 未处理的消息类型: {}", message.getClass().getSimpleName());
    }

    // 生命周期方法
    protected void preStart() {}
    protected void postStop() {}

    /**
     * Actor启动时调用（可选）
     */
    protected void onStart() {
        log.info("Actor启动: {}", getSelf().path());
    }

    /**
     * Actor停止时调用（可选）
     */
    protected void onStop() {
        log.info("Actor停止: {}", getSelf().path());
    }

    /**
     * 处理异常（可选）
     */
    protected void onError(Throwable error, Object message) {
        log.error("[{}] 处理消息时发生异常: {}", getSelf().path(), message, error);
    }

    // Getter/Setter
    public void setSelf(ActorRef self) {
        this.self = self;
    }

    public void setSystem(ActorSystem system) {
        this.system = system;
    }

    public ActorRef getSelf() {
        return self;
    }

    public ActorRef getSender() {
        return sender;
    }

    public ActorSystem getSystem() {
        return system;
    }

    // 接收构建器
    public class ReceiveBuilder {
        public <T> ReceiveBuilder match(Class<T> messageType, Consumer<T> handler) {
            messageHandlers.put(messageType, (Consumer<Object>) handler);
            return this;
        }

        // 添加 matchAny 方法
        public ReceiveBuilder matchAny(Consumer<Object> handler) {
            messageHandlers.put(Object.class, handler);
            return this;
        }
    }

    // 内部消息包装类
    private static class ActorMessage implements Serializable {
        final Object message;
        final ActorRef sender;

        ActorMessage(Object message, ActorRef sender) {
            this.message = message;
            this.sender = sender;
        }
    }

    // 内置消息类型
    public static class Heartbeat implements Serializable {
        private static final long serialVersionUID = 1L;
        public final String from;
        public final long timestamp;

        public Heartbeat(String from) {
            this.from = from;
            this.timestamp = System.currentTimeMillis();
        }
    }
}
